package com.example.im.infra.executor.send;

import com.example.im.endpoint.WebSocketEndpoint;
import com.example.im.infra.constant.ImConstants;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskExecutor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

/**
 * @author PC
 * 默认执行
 */
@Component
public class DefaultSendExecutor extends AbstractBaseSendExecutor {
    private final static Logger logger = LoggerFactory.getLogger(DefaultSendExecutor.class);

    private static TaskExecutor taskExecutor;

    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    public void setTaskExecutor(TaskExecutor taskExecutor) {
        DefaultSendExecutor.taskExecutor = taskExecutor;
    }

    @Autowired
    public void setRedisTemplate(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @Override
    public String getCommunicationType() {
        return ImConstants.CommunicationType.DEFAULT;
    }

    @Override
    public void sendToUser(String sendUserName, String message) {
        List<String> receiverNameList = getReceiverName(sendUserName, message);
        CountDownLatch countDownLatch = new CountDownLatch(receiverNameList.size());
        Set<String> notOnlineReceiverSet = ConcurrentHashMap.newKeySet();
        Set<String> finalNotOnlineReceiverSet = notOnlineReceiverSet;
        receiverNameList.forEach(receiverName -> taskExecutor.execute(() -> {
                            try {
                                this.sendMessage(receiverName, message, finalNotOnlineReceiverSet);
                            } catch (IOException ioException) {
                                logger.error("send error:" + ioException);
                            } finally {
                                countDownLatch.countDown();
                            }
                        }
                )
        );
        try {
            countDownLatch.await();
        } catch (InterruptedException interruptedException) {
            logger.error("error.countDownLatch.await");
        }
        notOnlineReceiverSet = notOnlineReceiverSet.stream().filter(StringUtils::isNotEmpty).collect(Collectors.toSet());
        if (CollectionUtils.isNotEmpty(notOnlineReceiverSet)) {
            logger.info("not online number is " + notOnlineReceiverSet.size());
            logger.info("The user : {} is not online", String.join(",", notOnlineReceiverSet));
        }
    }

    @Override
    public void sendToAll(String sendUserName, String message) {
        if (StringUtils.isBlank(sendUserName)) {
            logger.warn("send user name is blank");
        }
        for (Map.Entry<Integer, WebSocketEndpoint> webSocketEndpointEntry : WebSocketEndpoint.WEB_SOCKET_ENDPOINT_MAP.entrySet()) {
            taskExecutor.execute(() -> {
                        if (webSocketProperties.getReceiverExcludesHimselfFlag() && Objects.equals(sendUserName.hashCode(), webSocketEndpointEntry.getKey())) {
                            return;
                        }
                        try {
                            webSocketEndpointEntry.getValue().getSession().getBasicRemote()
                                    .sendText(generatorMessage(message));
                        } catch (IOException ioException) {
                            logger.error("send error:" + ioException);
                        }
                    }
            );
        }
    }

    private void sendMessage(String receiverName, String message, Set<String> finalNotOnlineReceiverSet) throws IOException {
        int receiverNameHashCode = receiverName.hashCode();
        if (userOnlineInServiceFlag(receiverNameHashCode)) {
            WebSocketEndpoint.WEB_SOCKET_ENDPOINT_MAP.get(receiverNameHashCode).getSession().getBasicRemote()
                    .sendText(generatorMessage(message));
        } else {
            handleNotOnline(receiverNameHashCode, receiverName, finalNotOnlineReceiverSet);
        }
    }

    /**
     * 当前服务用户是否在线
     *
     * @param hashCode 用户名的hashCode
     * @return 用户是否在线
     */
    private Boolean userOnlineInServiceFlag(int hashCode) {
        return WebSocketEndpoint.WEB_SOCKET_ENDPOINT_MAP.containsKey(hashCode);
    }

    /**
     * 不在线处理逻辑
     *
     * @param hashCode                  用户名hashCode
     * @param receiverName              用户名
     * @param finalNotOnlineReceiverSet 不在线用户名单
     */
    private void handleNotOnline(int hashCode, String receiverName, Set<String> finalNotOnlineReceiverSet) {
        if (Boolean.TRUE.equals(redisTemplate.opsForValue().getBit(WebSocketEndpoint.ONLINE_USER, hashCode))) {
            logger.debug(receiverName + " is not in this service");
        } else {
            finalNotOnlineReceiverSet.add(receiverName);
        }
    }
}
